package org.dingfugui.triple.link.zk;

import org.apache.zookeeper.*;
import org.dingfugui.triple.common.Logger;
import org.dingfugui.triple.common.LoggerFactory;

import java.io.Closeable;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * @author fugui.ding at 2015/6/26
 */
public class ZookeeperWatcher implements Watcher, Closeable {

    private final Logger logger = LoggerFactory.getLogger(getClass());
    private final String root = "/triple/service";
    private final String ip;
    private final int port;
    private final String appName;
    private final String path;
    private final ServerChangeRollback rollback;
    private ZooKeeper zk;

    public ZookeeperWatcher(String ip, int port, String appName, ServerChangeRollback rollback) throws IOException {
        this.ip = ip;
        this.port = port;
        this.appName = appName;
        this.rollback = rollback;
        this.path = root + "/" + appName;
        connect();
        listen();
    }

    private void connect() throws IOException {
        //连接zk
        final CountDownLatch lock = new CountDownLatch(1);
        zk = new ZooKeeper(ip + ":" + port, 10000, event -> {
            if (event.getState().equals(Event.KeeperState.SyncConnected)) {
                lock.countDown();
            }
        });
        try {
            lock.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        logger.info("zk连接成功");
    }

    public void listen() {
        try {
            //创建root
            if (zk.exists(path, false) == null) {
                zk.create(path, null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }
            //监控服务变动
            zk.getChildren(path, this);
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        logger.debug("zk正在监听");
    }

    public List<RemoteServerMetaData> getServiceList() {
        try {
            List<String> list = zk.getChildren(root, false);
            return parse(list);
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return Collections.emptyList();
    }

    @Override
    public void process(WatchedEvent event) {
        if (Event.EventType.NodeChildrenChanged.equals(event.getState())) {
            //子节点变动
            try {
                List<String> list = zk.getChildren(root, false);
                List<RemoteServerMetaData> servers = parse(list);
                rollback.call(servers);
            } catch (KeeperException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        try {
            //watch是一次性的，需重新注册
            zk.getChildren(path, this);
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    private List<RemoteServerMetaData> parse(List<String> list) throws KeeperException, InterruptedException {
        List<RemoteServerMetaData> result = new ArrayList<>(list.size());
        for (String child : list) {
            byte[] data = zk.getData(path + "/" + child, false, null);
            String url = new String(data);

            RemoteServerMetaData s = new RemoteServerMetaData();
            s.setAppName(appName);
            s.setUrl(url);
            result.add(s);
        }
        return result;
    }

    @Override
    public void close() {
        try {
            zk.close();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
